package com.action


import java.sql.Timestamp
import java.util.{Timer, TimerTask}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.functions.col
import org.apache.spark.sql.{DataFrame, Row, SaveMode, SparkSession}

import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet}
import java.util.Properties

object CountTopAndSaveForObject {
  // 创建 SparkSession
  val spark = SparkSession.builder()
    .appName("als-demo")
    .master("local[*]")
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    .getOrCreate()

  val timeinterval =  5* 60 * 1000

  // 从 MySQL 数据库读取数据
  val props = new Properties()
  props.setProperty("user", "root")
  props.setProperty("password", "123456")
  val jdbcUrl = "jdbc:mysql://localhost:3306/hf-test"

  def getCommDataframe(startTime: Timestamp): RDD[(Long, Long, String, Int,Timestamp)] = {

    val query =
      s"""
         |(SELECT user_id as userId, object_id as objectId, object_class as actionClass, action, create_time as createTime
         | FROM user_action_objext_log
         | WHERE create_time >= '${startTime.toString}'
         |) AS user_action_objext_log_tmp
           """.stripMargin
//    print("query:",query)

    val userId_trendId: DataFrame = spark.read.jdbc(jdbcUrl, query, props)

    // 将action列转换为Int类型
    val userId_trendIdWithIntAction: DataFrame = userId_trendId.withColumn("action", col("action").cast("int"))


    // 展示读取的数据，检查是否正确
//    println("DataFrame Schema:")
//    userId_trendIdWithIntAction.printSchema()

//    println("Sample Data:")
//    userId_trendIdWithIntAction.show()

    val rating: RDD[(Long, Long, String, Int,Timestamp)] = userId_trendIdWithIntAction.rdd.map {
      case Row(userId: Long, objectId: Long , actionClass: String, action: Int, createTime:Timestamp) => (userId, objectId, actionClass, action, createTime)
    }


    rating
  }


  // 定义一个方法来合并所有行为的 RDD
  def mergeBehaviorDataframes(): RDD[(String, (Long, Long, String, Int,Timestamp))]  = {
    // 计算当前时间往前推10分钟的时间戳
    val currentTime = new Timestamp(System.currentTimeMillis())
    val tenMinutesAgo = new Timestamp(currentTime.getTime() - (timeinterval))
    val commRDD = getCommDataframe(tenMinutesAgo)

    //
    val allRDD = commRDD

    // 按 trendId 进行分组
    val groupedRDD: RDD[(String, (Long, Long, String, Int,Timestamp))] = allRDD.map {
      case (userId, objectId, actionClass, action, createTime) => (actionClass,(userId, objectId, actionClass, action, createTime))
    }
    groupedRDD
  }

  def KeyByObjectClass(): RDD[(String, Iterable[(Long, Long, String, Int, Timestamp)])] = {
    val groupedRDD = mergeBehaviorDataframes()

    // 使用 groupByKey 方法根据 actionClass 进行分组
    val grouped: RDD[(String, Iterable[(Long, Long, String, Int, Timestamp)])] = groupedRDD.groupByKey()
    grouped
  }

  def countLikeAndSave(groupedRDD:RDD[(Long,(Long, Long,  Int, Timestamp))]): RDD[(Long, (Int, Int, Int))] = {

    // 对每组的数据进行分组和统计
    val countedRDD: RDD[(Long, (Int, Int, Int))] = groupedRDD.groupByKey().mapValues { iter =>
      var topCount = 0
      var commCount = 0
      var saveCount = 0

      iter.foreach {
        case (_, _, action, _) =>
          if (action == 2) topCount += 1
          else if (action == 22) topCount -= 1
          else if (action == 3) saveCount -= 1
          else if (action == 33) saveCount -= 1
          else if (action == 5) commCount -= 1
          else if (action == 55) commCount -= 1

      }
      (topCount, commCount, saveCount)
    }
    countedRDD
  }

  def countTop(groupedRDD: RDD[(Long, (Long, Long, Int, Timestamp))]): RDD[(Long, (Int))] = {

    // 对每组的数据进行分组和统计
    val countedRDD: RDD[(Long, (Int))] = groupedRDD.groupByKey().mapValues { iter =>
      var topCount = 0

      iter.foreach {
        case (_, _, action, _) =>
          if (action == 2) topCount += 1
          else if (action == 22) topCount -= 1

      }
      (topCount)
    }
    countedRDD
  }



  def getDataByObjectClass(grouped: RDD[(String, Iterable[(Long, Long, String, Int, Timestamp)])], objectClass: String): RDD[(Long, (Long, Long, Int, Timestamp))] = {
    // 根据 objectClass 进行过滤和转换
    val filteredRDD: RDD[(Long, Long, String, Int, Timestamp)] = grouped
      .filter { case (key, _) => key == objectClass }
      .flatMapValues(iterable => iterable) // 展开每个键值对的 Iterable
      .values // 只保留值部分

    // 对保留的值部分进行映射，提取需要的字段并返回新的 RDD
    val resultRDD: RDD[(Long, (Long, Long, Int, Timestamp))] = filteredRDD
      .map {
        case (userId, objectId, _, action, createTime) => (objectId, (userId, objectId, action, createTime))
      }

    resultRDD
  }

  // 定义方法来更新 trend 表中的数据
  def updateTrendTable(resultRDD: RDD[(Long, (Int, Int, Int))]): Unit = {
    // MySQL 连接配置
    val jdbcUrl = "jdbc:mysql://localhost:3306/hf-test"
    val user = "root"
    val password = "123456"

    // 将结果 RDD 转换为需要的 Map 数据结构
    val trendUpdates: Map[Long, (Int, Int, Int)] = resultRDD.collect().toMap.view.mapValues {
      case (newTopCount, newComCount, newSaveCount) => (newTopCount, newComCount, newSaveCount)
    }.toMap

    // 调用 updateTrendData2 方法执行更新
    updateTrendData2(jdbcUrl, user, password, trendUpdates)
  }

  // 定义方法来更新 trend 表中的数据
  def updateEvaluate(resultRDD: RDD[(Long, (Int))]): Unit = {
    // MySQL 连接配置
    val jdbcUrl = "jdbc:mysql://localhost:3306/hf-test"
    val user = "root"
    val password = "123456"

    // 将结果 RDD 转换为需要的 Map 数据结构
    val evaluateUpdates: Map[Long, (Int)] = resultRDD.collect().toMap.view.mapValues {
      case (newTopCount) => (newTopCount)
    }.toMap


    updateEvaluateData2(jdbcUrl, user, password, evaluateUpdates)
  }

  def updateComment(resultRDD: RDD[(Long, (Int))]): Unit = {
    // MySQL 连接配置
    val jdbcUrl = "jdbc:mysql://localhost:3306/hf-test"
    val user = "root"
    val password = "123456"

    // 将结果 RDD 转换为需要的 Map 数据结构
    val commentUpdates: Map[Long, (Int)] = resultRDD.collect().toMap.view.mapValues {
      case (newTopCount) => (newTopCount)
    }.toMap


    updateCommentData2(jdbcUrl, user, password, commentUpdates)
  }

  // 修改Trend表格数据
  def updateTrendData2(url: String, user: String, password: String, trendUpdates: Map[Long, (Int, Int, Int)]): Unit = {
    var connection: Connection = null
    var preparedStatement: PreparedStatement = null
    var resultSet: ResultSet = null

    try {
      // 连接数据库
      connection = DriverManager.getConnection(url, user, password)

      // 设置自动提交为 false，以便手动控制事务
      connection.setAutoCommit(false)

      // 遍历需要更新的博客数据
      trendUpdates.foreach { case (trendId, (newTopCount, newComCount, newSaveCount)) =>
        // 查询原始数据库中对应 trendId 的值
        val querySql =
          s"""
             |SELECT top_count, com_count, save_count
             |FROM trend
             |WHERE trend_id = ?
             |""".stripMargin

        preparedStatement = connection.prepareStatement(querySql)
        preparedStatement.setLong(1, trendId)
        resultSet = preparedStatement.executeQuery()

        if (resultSet.next()) {
          // 获取原始数据库中的值
          val originalTopCount = resultSet.getInt("top_count")
          val originalComCount = resultSet.getInt("com_count")
          val originalSaveCount = resultSet.getInt("save_count")

          // 计算新值
          val updatedTopCount = originalTopCount + newTopCount
          val updatedComCount = originalComCount + newComCount
          val updatedSaveCount = originalSaveCount + newSaveCount

          // 构建更新 SQL
          val updateSql =
            s"""
               |UPDATE trend
               |SET top_count = ?,
               |    com_count = ?,
               |    save_count = ?
               |WHERE trend_id = ?
               |""".stripMargin

          // 执行更新
          preparedStatement = connection.prepareStatement(updateSql)
          preparedStatement.setInt(1, updatedTopCount)
          preparedStatement.setInt(2, updatedComCount)
          preparedStatement.setInt(3, updatedSaveCount)
          preparedStatement.setLong(4, trendId)
          preparedStatement.executeUpdate()
        }
      }

      // 手动提交事务
      connection.commit()

      println("Trend update successful!")

    } catch {
      case e: Exception =>
        // 发生异常时回滚事务
        if (connection != null) connection.rollback()
        println("Error occurred while updating trend data: " + e.getMessage)
    } finally {
      // 关闭连接和 preparedStatement
      if (resultSet != null) resultSet.close()
      if (preparedStatement != null) preparedStatement.close()
      if (connection != null) connection.close()
    }
  }
  // 修改Evaluate表格数据
  def updateEvaluateData2(url: String, user: String, password: String, trendUpdates: Map[Long, (Int)]): Unit = {
    var connection: Connection = null
    var preparedStatement: PreparedStatement = null
    var resultSet: ResultSet = null

    try {
      // 连接数据库
      connection = DriverManager.getConnection(url, user, password)

      // 设置自动提交为 false，以便手动控制事务
      connection.setAutoCommit(false)

      // 遍历需要更新的博客数据
      trendUpdates.foreach { case (evaluateId, (newTopCount)) =>
        // 查询原始数据库中对应 trendId 的值
        val querySql =
          s"""
             |SELECT useful
             |FROM evaluates
             |WHERE evaluate_id = ?
             |""".stripMargin

        preparedStatement = connection.prepareStatement(querySql)
        preparedStatement.setLong(1, evaluateId)
        resultSet = preparedStatement.executeQuery()

        if (resultSet.next()) {
          // 获取原始数据库中的值
          val originalTopCount = resultSet.getInt("useful")

          // 计算新值
          val updatedTopCount = originalTopCount + newTopCount


          // 构建更新 SQL
          val updateSql =
            s"""
               |UPDATE evaluates
               |SET useful = ?
               |WHERE evaluate_id = ?
               |""".stripMargin

          // 执行更新
          preparedStatement = connection.prepareStatement(updateSql)
          preparedStatement.setInt(1, updatedTopCount)
          preparedStatement.setLong(2, evaluateId)
          preparedStatement.executeUpdate()
        }
      }
      // 手动提交事务
      connection.commit()

      println("Evaluate update successful!")

    } catch {
      case e: Exception =>
        // 发生异常时回滚事务
        if (connection != null) connection.rollback()
        println("Error occurred while updating trend data: " + e.getMessage)
    } finally {
      // 关闭连接和 preparedStatement
      if (resultSet != null) resultSet.close()
      if (preparedStatement != null) preparedStatement.close()
      if (connection != null) connection.close()
    }
  }

  // 修改Comment表格数据
  def updateCommentData2(url: String, user: String, password: String, trendUpdates: Map[Long, (Int)]): Unit = {
    var connection: Connection = null
    var preparedStatement: PreparedStatement = null
    var resultSet: ResultSet = null

    try {
      // 连接数据库
      connection = DriverManager.getConnection(url, user, password)

      // 设置自动提交为 false，以便手动控制事务
      connection.setAutoCommit(false)

      // 遍历需要更新的博客数据
      trendUpdates.foreach { case (commentId, (newTopCount)) =>
        // 查询原始数据库中对应 trendId 的值
        val querySql =
          s"""
             |SELECT top_count
             |FROM trend_comment
             |WHERE comment_id = ?
             |""".stripMargin

        preparedStatement = connection.prepareStatement(querySql)
        preparedStatement.setLong(1, commentId)
        resultSet = preparedStatement.executeQuery()

        if (resultSet.next()) {
          // 获取原始数据库中的值
          val originalTopCount = resultSet.getInt("top_count")

          // 计算新值
          val updatedTopCount = originalTopCount + newTopCount


          // 构建更新 SQL
          val updateSql =
            s"""
               |UPDATE trend_comment
               |SET top_count = ?
               |WHERE comment_id = ?
               |""".stripMargin

          // 执行更新
          preparedStatement = connection.prepareStatement(updateSql)
          preparedStatement.setInt(1, updatedTopCount)
          preparedStatement.setLong(2, commentId)
          preparedStatement.executeUpdate()
        }
      }
      // 手动提交事务
      connection.commit()

      println("Comment update successful!")

    } catch {
      case e: Exception =>
        // 发生异常时回滚事务
        if (connection != null) connection.rollback()
        println("Error occurred while updating trend data: " + e.getMessage)
    } finally {
      // 关闭连接和 preparedStatement
      if (resultSet != null) resultSet.close()
      if (preparedStatement != null) preparedStatement.close()
      if (connection != null) connection.close()
    }
  }


  def begin():Unit={
    val groupedRDD = KeyByObjectClass()

    // 获取特定 actionClass 的数据
    val trendRDD: RDD[(Long, (Long, Long, Int, Timestamp))] = getDataByObjectClass(groupedRDD, "trend")
    val evaluateRDD: RDD[(Long, (Long, Long, Int, Timestamp))] = getDataByObjectClass(groupedRDD, "evaluate")
    val commentRDD: RDD[(Long, (Long, Long, Int, Timestamp))] = getDataByObjectClass(groupedRDD, "comment")


    // 示例用法：统计每个博客的点赞数量、评论数量和收藏数量
    val resultTrendRDD: RDD[(Long, (Int, Int, Int))] = countLikeAndSave(trendRDD)

    val resultEvaluateRDD: RDD[(Long, (Int))] = countTop(evaluateRDD)

    val resultCommentRDD: RDD[(Long, (Int))] = countTop(commentRDD)

    // 检查是否有新数据生成，如果没有则不执行后续更新操作
    if (resultTrendRDD.isEmpty()) {
      println("No new trend data generated in the last 5 minutes. Skipping update.")
    } else {
      // 更新 trend 表中的数据
      updateTrendTable(resultTrendRDD)
//      println("有数据")
    }

    // 检查是否有新数据生成，如果没有则不执行后续更新操作
    if (resultEvaluateRDD.isEmpty()) {
      println("No new evaluate data generated in the last 5 minutes. Skipping update.")
    } else {
      // 更新 trend 表中的数据
      updateEvaluate(resultEvaluateRDD)
//      println("有数据")
    }

    // 检查是否有新数据生成，如果没有则不执行后续更新操作
    if (resultCommentRDD.isEmpty()) {
      println("No new comment data generated in the last 5 minutes. Skipping update.")
    } else {
      // 更新 trend 表中的数据
      updateComment(resultCommentRDD)
//      println("有数据")
    }




  }

  def main(args: Array[String]): Unit = {

    // 创建定时器对象
    val timer = new Timer()

    // 定义定时任务
    val task = new TimerTask {
      def run(): Unit = {
        // 在这里执行你想要定时执行的任务
        println("Executing task...")

        // 调用你的需要执行的函数
        begin()

        // 这里是示例代码，可以替换为你的逻辑

        println("Task completed.")
      }
    }


    // 每隔1分钟执行一次任务，初始延迟为0，间隔为1分钟（60 * 1000 毫秒）
    timer.schedule(task, 0, timeinterval)

    // 运行一段时间后取消定时器（模拟程序运行结束）
    Thread.sleep(1 * 60 * 1000) // 等待2分钟
    timer.cancel() // 取消定时任务
    println("Timer canceled.")


  }

}
